package org.pzy.pulsar;

import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.MessageIdImpl;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author pzy
 */
public class ReadDemo {
    public static void main(String[] args) throws PulsarClientException, ExecutionException, InterruptedException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.56.101:6650")
                .build();
        Reader<byte[]> reader = client.newReader().topic("topic-20240320")
                .startMessageId(MessageId.earliest).create();
        Reader<byte[]> reader2 = client.newReader().topic("topic-20240320")
                .startMessageId(new MessageIdImpl(9, 11, -1)).create();
        while (true) {
            Message<byte[]> msg = reader.readNext();
            System.out.println("从最开始开始读取 Message Got: " + new String(msg.getValue()) + ", messageId=" + msg.getMessageId());
            Message<byte[]> msg2 = reader2.readNext();
            System.out.println("从指定位置开始读取 Message Got: " + new String(msg2.getValue()) + ", messageId=" + msg2.getMessageId());
        }

    }
}
